/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.xhy.sniffer.datacarrier.consumer;


import org.xhy.sniffer.datacarrier.buffer.QueueBuffer;

import java.util.ArrayList;
import java.util.List;

public class ConsumerThread<T> extends Thread {
    private volatile boolean running; // 表示线程是否在运行，volatile 关键字保证线程安全的读写
    private IConsumer<T> consumer; // 消费者接口，用于处理数据的消费
    private List<DataSource> dataSources; // 数据源列表，每个数据源对应一个缓冲区，就是个 buffer
    private long consumeCycle; // 没有获取数据，线程 sleep 时间

    // 构造函数，初始化线程名称、消费者实例和消费周期
    ConsumerThread(String threadName, IConsumer<T> consumer, long consumeCycle) {
        super(threadName); // 设置线程名称
        this.consumer = consumer; // 赋值消费者
        running = false; // 初始化时线程未运行
        dataSources = new ArrayList<DataSource>(1); // 初始化数据源列表，初始容量为1
        this.consumeCycle = consumeCycle; // 设置消费周期
    }

    /**
     * 将整个缓冲区添加到消费队列中
     */
    void addDataSource(QueueBuffer<T> sourceBuffer) {
        // 将新的数据源缓冲区包装成 DataSource 并添加到数据源列表中
        this.dataSources.add(new DataSource(sourceBuffer));
    }

    @Override
    public void run() {
        running = true; // 设置线程为运行状态

        final List<T> consumeList = new ArrayList<T>(1500); // 用于存放消费的数据，初始容量为1500
        while (running) { // 当线程处于运行状态时，不断尝试消费数据
            if (!consume(consumeList)) { // 如果没有数据可消费
                try {
                    Thread.sleep(consumeCycle); // 线程休眠指定的周期
                } catch (InterruptedException e) {
                    // 捕获中断异常，但不处理
                }
            }
        }

        // 当线程准备停止时，最后一次尝试消费数据
        consume(consumeList);

        consumer.onExit(); // 调用消费者的退出方法，执行退出逻辑
    }

    /**
     * 从数据源中获取数据并消费
     *
     * @param consumeList 用于存储从数据源获取到的数据
     * @return 如果有数据被消费，返回 true；否则返回 false
     */
    private boolean consume(List<T> consumeList) {
        for (DataSource dataSource : dataSources) {
            // 从每个数据源中获取数据并存入 consumeList
            dataSource.obtain(consumeList);
        }

        if (!consumeList.isEmpty()) { // 如果获取到了数据
            try {
                consumer.consume(consumeList); // 调用消费者的 consume 方法处理数据
            } catch (Throwable t) {
                consumer.onError(consumeList, t); // 如果消费过程中发生异常，调用消费者的 onError 方法处理
            } finally {
                consumeList.clear(); // 无论成功或失败，最后清空 consumeList
            }
            return true; // 返回 true，表示成功消费了数据
        }

        consumer.nothingToConsume(); // 如果没有数据可消费，调用消费者的 nothingToConsume 方法
        return false; // 返回 false，表示没有数据被消费
    }

    /**
     * 关闭消费者线程
     */
    void shutdown() {
        running = false; // 设置运行标志为 false，线程将停止
    }

    /**
     * 内部类，用于封装数据源
     */
    class DataSource {
        private QueueBuffer<T> sourceBuffer; // 数据源对应的缓冲区

        // 构造函数，接收一个缓冲区
        DataSource(QueueBuffer<T> sourceBuffer) {
            this.sourceBuffer = sourceBuffer;
        }

        /**
         * 从数据源中获取数据并添加到 consumeList 中
         */
        void obtain(List<T> consumeList) {
            sourceBuffer.obtain(consumeList); // 调用缓冲区的 obtain 方法将数据获取到 consumeList 中
        }
    }
}

